﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Abp.Dependency;
using RabbitMQ.GNClient;
using RabbitMQ.GNClient.Events;
using GN.Pay.Core.Repositories;
using GN.Pay.Core.Entities;
using GN.Pay.Core.Values.Channel;
using Abp.Domain.Entities;
using GN.Pay.TradeChannel;

namespace GN.Pay.Core.Tasks.Queue.Impl
{
    /// <summary>
    /// 消息重置队列
    /// </summary>
    class ConsumerResetQueue : AConsumerQueue<QueueResetMessage>, IConsumerResetQueue
    {
        public ConsumerResetQueue(IIocManager iocManager)
            : base(QueueType.QueueReset, iocManager)
        {
        }

        /// <summary>
        /// 
        /// </summary>
        protected override ushort MaxPrefetchCount
        {
            get
            {
                return 3;
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="eventConsumer"></param>
        /// <param name="data"></param>
        /// <param name="e"></param>
        /// <returns></returns>
        protected override async Task Consumer(IEventingConsumer eventConsumer, QueueResetMessage data, DeliverEventArgs e)
        {
            await Task.Run(() =>
            {
                try
                {
                    PublishNotifyReset();
                    eventConsumer.Channel.Ack(e.DeliveryTag);
                }
                catch (Exception err)
                {
                    try
                    {
                        Logger.Error("消费重置队列异常:", err);
                        this.DelayConsumer(eventConsumer, data, e, new TimeSpan(0, 0, 30));                     
                    }
                    catch { }
                }
            });
        }

        private const int PageSize = 1000;       

        private void PublishNotifyReset()
        {
            var queue = IocManager.Resolve<IQueueManager>();
            var repository = IocManager.Resolve<IQueueErrorMessageRepository>();
            Exception channelErr = null;
            this.UpdateUnitOfWork(() =>
            {
                long lastId = 0;
                var items = new List<QueueErrorMessage>();
                bool isExit = false;
                do
                {
                    items = repository.GetAll().Where(s => s.Id > lastId).OrderBy(s => s.Id).Take(PageSize).ToList();
                    foreach (var item in items)
                    {
                        try
                        {
                            bool isDelete = true;
                            switch (item.MessageType)
                            {
                                case QueueResetMessage.ResetType_ChannelPayNotify:
                                    PublishPayNotifyReset(queue, item);
                                    break;
                                case QueueResetMessage.ResetType_ChannelRefundNotify:
                                    PublishRefundNotifyReset(queue, item);
                                    break;
                                case QueueResetMessage.ResetType_ChannelTransfersNotify:
                                    PublishTransfersNotifyReset(queue, item);
                                    break;
                                default:
                                    Logger.ErrorFormat("消费重置队列无法处理类别:({0})", item.MessageType);
                                    isDelete = false;
                                    break;                                  
                            }
                            if (isDelete)
                            {
                                repository.Delete(item);
                            }                           
                        }
                        catch (Exception err)
                        {
                            channelErr = err;
                            isExit = true;
                            break;
                        }
                    }
                    if (isExit)
                    {
                        break;
                    }
                    if (items.Count > 0)
                    {
                        lastId = items[items.Count - 1].Id;
                    }
                } while (items.Count >= PageSize);
            });
            if (channelErr != null)
            {
                throw channelErr;
            }
        }

        private void PublishPayNotifyReset(IQueueManager queue, QueueErrorMessage queueMessage)
        {
            var result = queueMessage.CreateOrderQueryResult();
            var msg = new ObjectMessage(result);
            msg.MessageId = result.MessageSign(Consts.Start.MessageSign);
            queue.Publish(QueueType.ChannelPayNotify, msg);
        }

        private void PublishRefundNotifyReset(IQueueManager queue, QueueErrorMessage queueMessage)
        {
            var result = queueMessage.CreateRefundApplyResult();
            var msg = new ObjectMessage(result);
            msg.MessageId = result.MessageSign(Consts.Start.MessageSign);
            queue.Publish(QueueType.ChannelRefundNotify, msg);
        }

        private void PublishTransfersNotifyReset(IQueueManager queue, QueueErrorMessage queueMessage)
        {
            var result = queueMessage.CreateTransfersApplyResult();
            var msg = new ObjectMessage(result);
            msg.MessageId = result.MessageSign(Consts.Start.MessageSign);
            queue.Publish(QueueType.ChannelTransfersNotify, msg);
        }
    }
}
